
#include "thread_pool.h"

#include <cassert>

#if defined(__GNUC__) || defined(__MINGW__) || defined(__MINGW32__)
    #pragma GCC diagnostic push
    #pragma GCC diagnostic ignored "-Wold-style-cast"
#endif


#if dLOGGING_THREAD_POOL
    #include "thread_cout.h"
    #include <iostream>
    using std::cout;
    using std::endl;
    #define pcout(msg) thread_cout(msg)
#else
    namespace{ inline void dummy(){} }//namespace
    #define pcout(msg) dummy()
#endif


namespace
{
    size_t Count(const unsigned num_threads)
    {
        return num_threads==0?
            std::thread::hardware_concurrency():
            num_threads;
    }

}//namespace

ThreadPool::ThreadPool(const unsigned num_threads)
    :mMutexQueue()
    ,mMutexStop()
    ,mThreads()
    ,mCondition()
//    ,mActive(0)
//    ,mSleeping( static_cast<char>( Count(num_threads) ) )
    ,mCountTask()
    ,mStopped(false)
    ,mTaskStop()
    ,mTaskQueue()
{
    const auto num = Count(num_threads);

    #ifdef DEBUG
        const bool ERROR_MINIMUM_THREADS_MUST_BE_TWO = num<2;
        assert(!ERROR_MINIMUM_THREADS_MUST_BE_TWO);
    #endif

    mThreads.reserve(num);

    pcout(cout<< "[ POOL ] launch with "<<num <<" threads"<< endl);

    for(size_t n=0; n<num; ++n)
        mThreads.emplace_back(
            [this](){ this->ThreadFunction(); }
        );
}




void ThreadPool::ThreadFunction()
{
    #if dLOGGING_THREAD_POOL
    const auto id = std::this_thread::get_id();
    pcout(cout<< "[THREAD]["<<id<<"] started" << endl);
    #endif

    Task task;

    for( ;; )
    {
        {
            std::unique_lock<std::mutex>
                locker(mMutexStop);

            if( mCountTask == 0 && !mStopped )
                pcout(cout<< "[THREAD]["<<id<<"] sleeping" << endl),
                mCondition.wait(locker);
        }
        pcout(cout<< "[THREAD]["<<id<<"] wake up!" << endl);

        mMutexQueue.lock();
        if( !mTaskQueue.empty() )
            task = mTaskQueue.front(),
            mTaskQueue.pop(),
            --mCountTask,
            mMutexQueue.unlock(),
            mCondition.notify_one(),
            pcout(cout<< "[THREAD]["<<id<<"] processing... " << endl),
            task();
        else
        {
            mMutexQueue.unlock();
            if(mStopped)
                break;
        }
    }

    pcout(cout<< "[THREAD]["<<id<<"] stopping..." << endl);
    if(mTaskStop)
        mTaskStop();
    pcout(cout<< "[THREAD]["<<id<<"] -----------------> stopped" << endl);
}

ThreadPool& ThreadPool::Start()
{ 
    #if dLOGGING_THREAD_POOL
        const auto id = std::this_thread::get_id();
        pcout(cout<< "[ POOL ]["<<id<<"][NEW TASK] adding... " << endl);
    #endif

    std::unique_lock<std::mutex> locker(mMutexStop);

    #if dLOGGING_THREAD_POOL
        pcout(cout<< "[ POOL ]["<<id<<"][NEW TASK] signal notify_one " << endl);
    #endif

    mCondition.notify_one();

    #if dLOGGING_THREAD_POOL
        pcout(cout<< "[ POOL ]["<<id<<"][NEW TASK] adding done!" << endl);
    #endif

    return *this;
}

void ThreadPool::Stop()
{
    std::unique_lock<std::mutex>
        locker(mMutexStop);

    mStopped = true;

    #if dLOGGING_THREAD_POOL
    const auto id = std::this_thread::get_id();
    pcout(cout<< "[ POOL ]["<<id<<"] stopped, signal notify_all: "<<(int)mSleeping<< endl);
    #endif

    mCondition.notify_all();
}

void ThreadPool::Wait()
{
    #if dLOGGING_THREAD_POOL
    const auto id = std::this_thread::get_id();
    pcout(cout<< "[ POOL ]["<<id<<"] wait all thread..." << endl);
    #endif

    for(auto& t: mThreads)
        t.join();
}


ThreadPool& ThreadPool::Get(const size_t num_threads)
{
    static ThreadPool p(num_threads);
    return p;
}


#if defined(__GNUC__) || defined(__MINGW__) || defined(__MINGW32__)
    #pragma GCC diagnostic pop
#endif

#undef pcout

